import { Snippet } from "@/components/code";
import { snippets } from "@/lib/generated/snippets";

import { Callout, Card, Cards, Steps, Tabs } from "nextra/components";
import UniversalTabs from "@/components/UniversalTabs";

## Introduction

Hatchet V1 introduces the ability to add conditions to tasks in your workflows that determine whether or not a task should be run, based on a number of conditions. Conditions unlock a number of new ways to solve problems with Hatchet, such as:

1. A workflow that reads a feature flag, and then decides how to progress based on its value. In this case, you'd have two tasks that use parent conditions, where one task runs if the flag value is e.g. `True`, while the other runs if it's `False`.
2. Any type of human-in-the-loop workflow, where you want to wait for a human to e.g. approve something before continuing the dag.

## Types of Conditions

There are three types of `Condition`s in Hatchet V1:

1. Sleep conditions, which sleep for a specified duration before continuing
2. Event conditions, which wait for an event (and optionally a CEL expression evaluated on the payload of that event) before deciding how to continue
3. Parent conditions, which wait for a parent task to complete and then decide how to progress based on its output.

## Or Groups

Conditions can also be combined using an `Or` operator into groups of conditions (called "or groups") where at least one must be satisfied in order for the group to evaluate to `True`. An "or group" behaves like a boolean `OR` operator, where the group evaluates to `True` if at least one of its conditions is `True`.

Or groups are an extremely powerful feature because they let you express arbitrarily complex sets of conditions in [conjunctive normal form](https://en.wikipedia.org/wiki/Conjunctive_normal_form) (CNF) for determining when your tasks should run and when they should not. As a simple example, consider the following conditions:

- **Condition A**: Checking if the output of a parent task is greater than 50
- **Condition B**: Sleeping for 30 seconds
- **Condition C**: Receiving the `payment:processed` event

You might want to progress in your workflow if A _or_ B and C. In this case, we can express this set of conditions in CNF as `A or B` AND `A or C` where both `A or B` and `A or C` are or groups.

## Usage

Conditions can be used at task _declaration_ time in three ways:

1. They can be used in a `wait_for` fashion, where a task will wait for the conditions to evaluate to `True` before being run.
2. They can be used in a `skip_if` fashion, where a task will be skipped if the conditions evaluate to `True`.
3. They can be used in a `cancel_if` fashion, where a task will be cancelled if the conditions evaluate to `True`.

### `wait_for`

Declaring a task with conditions to `wait_for` will cause the task to wait before starting for until its conditions evaluate to `True`. For instance, if you use `wait_for` with a 60 second sleep, the workflow will wait for 60 seconds before triggering the task. Similar, if the task is waiting for an event, it will wait until the event is fired before continuing.

### `skip_if`

Declaring a task with conditions to `skip_if` will cause the task to be skipped if the conditions evaluate to `True`. For instance, if you use a parent condition to check if the output of a parent task is equal to some value, the task will be skipped if that condition evaluates to `True`.

### `cancel_if`

Declaring a task with conditions to `cancel_if` will cause the task to be cancelled if the conditions evaluate to `True`. For instance, if you use a parent condition to check if the output of a parent task is equal to some value, the task will be cancelled if that condition evaluates to `True`.

<Callout type="warning">
  A task cancelled by a `cancel_if` operator will behave the same as any other
  cancellation in Hatchet, meaning that downstream tasks will be cancelled as
  well.
</Callout>

## Example Workflow

In this example, we're going to build the following workflow:

![Branching DAG Workflow](/branching-dag.png)

Note the branching logic (`left_branch` and `right_branch`), as well as the use of skips and waits.

To get started, let's declare the workflow.

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
    <Snippet src={snippets.python.conditions.worker.create_a_workflow} />
  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
    <Snippet
      src={
        snippets.typescript.dag_match_condition.complex_workflow
          .create_a_workflow
      }
    />
  </Tabs.Tab>
  <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.create_a_workflow} />
  </Tabs.Tab>
</UniversalTabs>

Next, we'll start adding tasks to our workflow. First, we'll add a basic task that outputs a random number:

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
    <Snippet src={snippets.python.conditions.worker.add_base_task} />
  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
    <Snippet
      src={
        snippets.typescript.dag_match_condition.complex_workflow.add_base_task
      }
    />
  </Tabs.Tab>
  <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.add_base_task} />
  </Tabs.Tab>
</UniversalTabs>

Next, we'll add a task to the workflow that's a child of the first task, but it has a `wait_for` condition that sleeps for 10 seconds.

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
    <Snippet src={snippets.python.conditions.worker.add_wait_for_sleep} />
  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
    <Snippet
      src={
        snippets.typescript.dag_match_condition.complex_workflow
          .add_wait_for_sleep
      }
    />
  </Tabs.Tab>
  <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.add_wait_for_sleep} />
  </Tabs.Tab>
</UniversalTabs>

This task will first wait for the parent task to complete, and then it'll sleep for 10 seconds before executing and returning another random number.

Next, we'll add a task that will be skipped on an event:

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
    <Snippet src={snippets.python.conditions.worker.add_skip_on_event} />
  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
    <Snippet
      src={
        snippets.typescript.dag_match_condition.complex_workflow
          .add_skip_on_event
      }
    />
  </Tabs.Tab>
  <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.add_skip_on_event} />
  </Tabs.Tab>
</UniversalTabs>

In this case, our task will wait for a 30 second sleep, and then it will be skipped if the `skip_on_event:skip` is fired.

Next, let's add some branching logic. Here we'll add two more tasks, a left and right branch.

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
    <Snippet src={snippets.python.conditions.worker.add_branching} />
  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
    <Snippet
      src={
        snippets.typescript.dag_match_condition.complex_workflow.add_branching
      }
    />
  </Tabs.Tab>
  <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.add_branching} />
  </Tabs.Tab>
</UniversalTabs>

These two tasks use the `ParentCondition` and `skip_if` together to check if the output of an upstream task was greater or less than `50`, respectively. Only one of the two tasks will run: whichever one's condition evaluates to `True`.

Next, we'll add a task that waits for an event:

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
    <Snippet src={snippets.python.conditions.worker.add_wait_for_event} />
  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
    <Snippet
      src={
        snippets.typescript.dag_match_condition.complex_workflow
          .add_wait_for_event
      }
    />
  </Tabs.Tab>
  <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.add_wait_for_event} />
  </Tabs.Tab>
</UniversalTabs>

And finally, we'll add the last task, which collects all of its parents and sums them up.

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">
<Snippet src={snippets.python.conditions.worker.add_sum}/>

Note that in this task, we rely on `ctx.was_skipped` to determine if a task was skipped.

  </Tabs.Tab>
  <Tabs.Tab title="Typescript">
<Snippet src={snippets.typescript.dag_match_condition.complex_workflow.add_sum}/>
  </Tabs.Tab>
    <Tabs.Tab title="Go">
    <Snippet src={snippets.go.conditions.main.add_sum}/>
  </Tabs.Tab>

</UniversalTabs>

This workflow demonstrates the power of the new conditional logic in Hatchet V1. You can now create complex workflows that are much more dynamic than workflows in the previous version of Hatchet, and do all of it declaratively (rather than, for example, by dynamically spawning child workflows based on conditions in the parent).
